package day1

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}

//Spark3.4.*目前仅支持来自java8，11

object RDD基础操作 {
  def main(args: Array[String]): Unit = {
    //定义hadoop内容
    System.setProperty("hadoop.home.dir", "D:\\新建文件夹\\HCIA初级\\hadoop")

    //    //配置Spark
    //    val conf = new SparkConf()
    //    //任务名称
    //    conf.setAppName("Basic")
    //    /* 设置Spark工作模式
    //     * local：本地工作模式，更使用学习Spark的操作处理，更方便调用，只适合学习使用
    //     * 面向生产环境：
    //     * Standalone：集群化环境
    //     * Yarn：Hadoop集群
    //     */
    //    conf.setMaster("local")

    //基于Scala中一切皆为对象，可以将配置Spark操作合并
    //    new SparkConf().setAppName("Basic").setMaster("local")

    //关闭日志
    Logger.getLogger("org").setLevel(Level.OFF)

    //创建一个Spark的任务入口：SparkContext()
    val sc = new SparkContext(new SparkConf().setAppName("Basic").setMaster("local"))

    //RDD的数据内容来自两个部分
    //1、读取来自外部的数据集内容
    val rdd_file = sc.textFile("D:\\新建文件夹\\data\\wordCount.txt")
    //2、在驱动程序中进行数据获取，并行化
    val arr = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
    val rdd = sc.parallelize(arr)

    /*
     * RDD是一个不可变类型数据
     * RDD算子一共有两种算子：转化(transformation)算子、行动(action)算子
     * * 转化(transformation)算子：由一个RDD生成另外一个RDD，转化操作针对于前一个RDD中的各个元素
     * * 行动(action)算子：计算各个算子转化操作
     * 转化(transformation)算子生成的内容一定是RDD类型，行动(action)算子一定不是RDD类型
     * 什么叫惰性计算？
     * * 因为Spark是基于内存进行处理操作，为了节约内存空间，避免被浪费
     * * 所以在遇到行动算子时才会运行
     * 对于Spark来说，当了解了完整的操作链之后，计算只需要用到的数据
     */

    //退出代码表示当前任务运行的状态：
    //0 ：当前任务运行成功；非0：当前任务失败

    //使用一个转化算子map生成一个新的RDD类型：rdd_map
    //转化算子map：读取前一个RDD的数据，通过计算放置在新的RDD上（前一个RDD、新RDD长度一致）
    val rdd_map = rdd.map(i => println("当前数值的乘积为：" + i * i))

    //执行算子：foreach()，遍历RDD并按照输出格式定义输出
    //    rdd_map.foreach(println)
    //执行算子：count(),直接读取RDD,返回RDD的长度
    //    rdd_map.count()
    //执行算子：first()：读取RDD的第一个数据
    //    rdd_map.first()

    //谱系图例子
    val arr1 = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
    val arr2 = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
    val rdd1 = sc.parallelize(arr1)
    val rdd2 = sc.parallelize(arr2)

    val rdd1_map = rdd1.map(i => println(s"当前为rdd1_map的数据：$i"))
    val rdd2_map = rdd2.map(i => println(s"当前为rdd2_map的数据：$i"))
    //转化算子：union()，将前一个RDD的数据与后一个RDD的数据合并，生成一个新的RDD
    val rdd1And2 = rdd1_map.union(rdd2_map)
    val count = rdd1And2.count()
    print(count)
    //通过谱系图例子，可以得到一个谱系图
    //谱系图的作用：通过表达依赖关系从而可以在出现父RDD丢失的情况下进行数据恢复

  }

}